package com.atguigu.flink.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.time.Duration;

/**
 * Created by Smexy on 2023/4/11
 *
 *
 *  默认是 对齐，是EOS。
 *   env.enableCheckpointing(2000, CheckpointingMode.AT_LEAST_ONCE); 对齐，是at least once
 *
 *   --------------------------------
 *      不对齐:  checkpointConfig.enableUnalignedCheckpoints(true);
 *          Unaligned checkpoints can only be used with checkpointing mode EXACTLY_ONCE
 *              不对齐，只有 EOS！
 *
 *
 *
 */
public class Demo13_CheckpointConfig
{
    public static void main(String[] args) {


        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //ck的设置
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        //设置为不对齐
        checkpointConfig.enableUnalignedCheckpoints(true);
        //设置ck超时时间
        checkpointConfig.setCheckpointTimeout(60000);
        //两次ck之间，至少要间隔多久
        checkpointConfig.setMinPauseBetweenCheckpoints(500);
        //当前允许同时进行ck的数量。限制流中barrier的数量。 不对齐，这个只能是1
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        //设置对齐方式下，ck的超时时间
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(2));
        //ck失败了20次，Job就停止
        checkpointConfig.setTolerableCheckpointFailureNumber(20);
        // checkpoints with Finished Tasks 最终检测点。如果要停止一个Job，会等Task完成了最近的一次CK后，再停止Job
        //默认情况，job被cancel了，就把ck的备份删除。 RETAIN_ON_CANCELLATION： Job被cacel后，ck依旧保留
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);


        env.setParallelism(2);

        //env.enableCheckpointing(2000, CheckpointingMode.AT_LEAST_ONCE);
        //env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.enableCheckpointing(2000);
        checkpointConfig.setCheckpointStorage("file:///d://tmp");

        env
            .socketTextStream("hadoop102", 8888)
            .map(new Demo1_State.MyMapFunction())
            .addSink(new SinkFunction<String>()
            {
                @Override
                public void invoke(String value, Context context) throws Exception {
                    if (value.contains("x")){
                        //模拟抛异常
                        throw new RuntimeException("出异常了...");
                    }
                    System.out.println(value);
                }
            });


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    //进行对每次输入的字符串进行累加拼接
    public static class MyMapFunction implements MapFunction<String,String>, CheckpointedFunction
    {


        private ListState<String> result;

        @Override
        public String map(String value) throws Exception {
            result.add(value);
            return result.get().toString();
        }


        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
        }

        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {

            OperatorStateStore operatorStateStore = context.getOperatorStateStore();

            ListStateDescriptor<String> listStateDescriptor = new ListStateDescriptor<>("strs", String.class);
            result = operatorStateStore.getListState(listStateDescriptor);
        }
    }
}
